package org.qiyu.live.msg.provider.consumer;

import com.alibaba.fastjson2.JSON;
import jakarta.annotation.Resource;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.dubbo.live.common.interfaces.topic.ImCoreServerProviderTopicNames;
import org.qiyu.live.framework.mq.starter.properties.RocketMQConsumerProperties;
import org.qiyu.live.im.dto.ImMsgBody;
import org.qiyu.live.msg.provider.consumer.Handler.MessageHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

/**
 * @Description: mq消费
 * @Author: 张广焱
 * @Date: 2024-05-03 下午11:11
 */
@Component
public class ImMsgConsumer implements InitializingBean {
    private static final Logger logger = LoggerFactory.getLogger(ImMsgConsumer.class);

    @Resource
    private RocketMQConsumerProperties rocketMQConsumerProperties;

    @Resource
    private MessageHandler messageHandler;

    /**
     * @Description: 在初始化bean的时候都会执行该方法。
     * @Author: 张广焱
     * @Date: 2024-05-03  23:12:10
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        // 创建一个消费者
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer();
        // 设置vip通道
        mqPushConsumer.setVipChannelEnabled(false);
        // 设置namesrv地址
        mqPushConsumer.setNamesrvAddr(rocketMQConsumerProperties.getNameSrv());
        // 设置消费者组
        mqPushConsumer.setConsumerGroup(rocketMQConsumerProperties.getGroupName() + "_" + ImMsgConsumer.class.getSimpleName());
        // 设置一次拉取10条消息
        mqPushConsumer.setConsumeMessageBatchMaxSize(10);
        // 设置消费位置
        mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // 监听im发送过来的业务消息topic
        mqPushConsumer.subscribe(ImCoreServerProviderTopicNames.QIYU_LIVE_IM_BIZ_MSG_TOPIC, "");
        mqPushConsumer.setMessageListener((MessageListenerConcurrently) (msgs, context) -> {
            msgs.forEach(msg -> {
                ImMsgBody imMsgBody = JSON.parseObject(new String(msg.getBody()), ImMsgBody.class);
                messageHandler.onMsgReceiver(imMsgBody);
            });

            // 处理消息成功
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        // 启动消费者
        mqPushConsumer.start();
        logger.info("mq消费组启动成功,namesrv is {}", rocketMQConsumerProperties.getNameSrv());
    }
}
